package cn.ac.iscas.service.transformer;

import cn.ac.iscas.base.utils.ParameterUtils;
import cn.ac.iscas.base.utils.SpringUtil;
import cn.ac.iscas.base.utils.TypeConvertUtil;
import cn.ac.iscas.domain.source.FlinkSourceColumn;
import cn.ac.iscas.service.queue.flink.FlinkTableQueueService;

import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author LJian
 * @version 1.0
 * @description: TODO
 * @date 2022/7/19 16:04
 */
public class BucketTransformer {
    private final static String ID = "bucket_trans";
    private final static String COLUMN_NAME_KEY = "column";
    private final static String LABEL_NAME_KEY = "label";
    private final static String BUCKET_NAME_KEY = "buckets";
    private final static String COLUMN_TYPE_KEY = "type";
    private String jobId;
    private String componentId;
    private String frontTableName;
    private List<FlinkSourceColumn> frontColumns;

    private String currentTableName;
    private List<FlinkSourceColumn> currentColumns;

    public BucketTransformer(String jobId, String taskId) {
        this.jobId = jobId;
        this.componentId = taskId;
    }

    public BucketTransformer(String jobId, String taskId, String frontTableName, List<FlinkSourceColumn> frontColumns) {
        this.jobId = jobId;
        this.componentId = taskId;
        this.frontTableName = frontTableName;
        this.frontColumns = frontColumns;
        this.currentColumns = frontColumns;
    }


    public BucketTransformer() {

    }

    //String sql = "CREATE VIEW middle_platform.db1.view3 AS (Select *, CASE " +
    //                " WHEN C16 < 50 then 'B01' " +
    //                " WHEN C16 >= 50 and C16 <250 then 'B02' " +
    //                " WHEN C16 >= 250 then 'B03' " +
    //                " END C200 from middle_platform.db1.view2 )" ";
    public String getSql(Map<String, Object> parameters) {
        //this.currentTableName = "bucket_" + UUID.randomUUID().toString().substring(0,6);
        this.currentTableName = String.format("%s.%s.%s_%s", jobId, jobId, ID, UUID.randomUUID().toString().substring(0,6));
        //String createViewSql = String.format("CREATE VIEW %s.%s.%s AS ", jobId, jobId, this.currentTableName);
        String createViewSql = String.format("CREATE VIEW %s AS ", this.currentTableName);
        Map<String, Map<String, Object>> buckets = ParameterUtils.getRequiredMap(parameters, BUCKET_NAME_KEY);
        FlinkSourceColumn label = ParameterUtils.getRequiredObj(parameters, LABEL_NAME_KEY, FlinkSourceColumn.class);
        FlinkSourceColumn column = ParameterUtils.getRequiredObj(parameters, COLUMN_NAME_KEY, FlinkSourceColumn.class);
        String columnName = column.getName();
        StringBuffer caseBuffer = new StringBuffer(String.format(" ( SELECT %s ,CASE", TypeConvertUtil.columnsToSelectString(this.frontColumns)));

        for (Map.Entry<String, Map<String, Object>> entry : buckets.entrySet()) {
            String bucketVal = entry.getKey();


            if (column.getType().equalsIgnoreCase("string")) {
                bucketVal = String.format("'%s'", bucketVal);
            }
            Map<String, Object> bucketInfo = entry.getValue();
            //" WHEN C16 < 50 then 'B01' "
            if (bucketInfo.get("lower") == null) {
                String operator = (boolean) bucketInfo.get("include_upper") ? "<=" : "<";

                caseBuffer.append(String.format(" WHEN %s %s %s THEN %s ", columnName, operator, bucketInfo.get("upper"), bucketVal));
                continue;
            }

            //" WHEN C16 >= 250 then 'B03' "
            if (bucketInfo.get("upper") == null) {
                String operator = (boolean) bucketInfo.get("include_lower") ? ">=" : ">";
                caseBuffer.append(String.format(" WHEN %s %s %s THEN %s ", columnName, operator, bucketInfo.get("lower"), bucketVal));
                continue;
            }

            //" WHEN C16 >= 50 and C16 <250 then 'B02' "
            String operatorLower = (boolean) bucketInfo.get("include_lower") ? ">=" : ">";
            String operatorUpper = (boolean) bucketInfo.get("include_upper") ? "<=" : "<";
            caseBuffer.append(String.format(" WHEN %s %s %s AND %s %s %s THEN %s ",
                    columnName, operatorLower, bucketInfo.get("lower"),
                    columnName, operatorUpper, bucketInfo.get("upper"),
                    bucketVal));
        }
        //" END C200 from middle_platform.db1.view2 )" "
        caseBuffer.append(String.format(" END %s FROM %s )", label.getName(), this.frontTableName));

        this.currentColumns.add(label);
        updateSource();
        createViewSql += caseBuffer.toString();
        return createViewSql;
    }

    public void updateSource() {
        FlinkTableQueueService flinkTableQueueService = SpringUtil.getBean(FlinkTableQueueService.class);
        Map<String, List<FlinkSourceColumn>> tableColumns = new ConcurrentHashMap<>(8);
        tableColumns.put(this.currentTableName, this.currentColumns);
        flinkTableQueueService.put(this.jobId, this.componentId, tableColumns);
    }
}
